package com.newbee.db;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class DBJob {
    public static void main(String[] args) {
        try {
            Configuration configuration = new Configuration();
            DBConfiguration.configureDB(configuration, "com.mysql.jdbc.Driver",
                    "jdbc:mysql://10.168.0.132:3306/unit?user=root&password=TrjX7IclV4@cMkSBT3Yg&useUnicode=true&characterEncoding=UTF8&useSSL=false", "root", "TrjX7IclV4@cMkSBT3Yg");
            Job job = Job.getInstance(configuration, "dbJob");
            final FileSystem fileSystem = FileSystem.get(new URI("/output"),
                    configuration);
            fileSystem.delete(new Path("/output"), true);
            job.setJarByClass(DBJob.class);
            job.setInputFormatClass(DBInputFormat.class);
            //Job job, Class inputClass, String tableName, String conditions, String orderBy, String... fieldNames
            DBInputFormat.setInput(job, Instance.class, "instance", "id<29", null, "id", "instance_content");
            // 当map输出类型和reduce输出类型一致时，可以不设置
            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(IntWritable.class);
            job.setMapOutputValueClass(Text.class);
            //指定不需要使用reduce，直接把map任务的输出写入到HDFS中
            job.setNumReduceTasks(0);
            FileOutputFormat.setOutputPath(job, new Path("/output"));
            job.waitForCompletion(true);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    public static class MyMapper extends
            Mapper<LongWritable, Instance, IntWritable, Text> {
        @Override
        protected void map(LongWritable key, Instance value, Context context) throws IOException, InterruptedException {
            System.out.println(value);
            context.write(new IntWritable(value.getId()), new Text(value.getInstanceContent()));
        }
    }
}
